package com.yyx.zookeeperlock.utils;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;

public class DistributedLock {
    private CuratorFramework client = null; //zk客戶端

    final static Logger log = LoggerFactory.getLogger(DistributedLock.class);


    public static final String ZOOKEEPER_SERVER="XXXXXXXXX";

    //用于挂起当前请求，并且等待上一个分布式锁释放
    private static CountDownLatch zkLocklatch = new CountDownLatch(1);

    //分布式锁的总节点名
    private static final String ZK_LOCK_PROJECT = "yyx-locks";

    //分布式锁节点
    private static final String DISTRIBUTED_LOCK = "distributed_lock";

    //构造函数
//    public DistributedLock(CuratorFramework client) {
//        this.client = client;
//    }


    public void init() {

        /**
         * curator链接zookeeper的策略:RetryNTimes
         * n：重试的次数
         * sleepMsBetweenRetries：每次重试间隔的时间
         */
        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);

        client = CuratorFrameworkFactory.builder()
                .connectString(ZOOKEEPER_SERVER)
                .sessionTimeoutMs(10000).retryPolicy(retryPolicy)
                .namespace("ZKLocks-Namespace").build();
        client.start();


        // 使用命名空间
        client = client.usingNamespace("ZKLocks-Namespace");

        /**
         * 创建zk锁的总节点，相当于idea的工作空间下的项目
         */

        try {
            if (client.checkExists().forPath("/" + ZK_LOCK_PROJECT) == null) {
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE);
            }
            //针对zk的分布式锁节点，创建对应的watcher事件监听
            addWatcherToLock("/" + ZK_LOCK_PROJECT);
            log.info("zookeeper客户端连接初始化成功...");
        } catch (Exception e) {
            e.printStackTrace();
            log.info("zookeeper客户端连接初始化失败...");
        }

    }


    /**
     * 获得分布式锁
     */
    public void getLock() {
        //使用死循环,当且仅当上一个锁释放并且当前请求活动锁成功后才会跳出
        while (true) {
            try {
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath("/" + ZK_LOCK_PROJECT + "/" + DISTRIBUTED_LOCK);
                log.info("获得分布式锁成功");
                return;   //如果锁的节点能被创建成功,则锁没有被占用
            } catch (Exception e) {
                log.info("获得分布式锁失败");
                try {
                    //如果没有获取到锁，需要重新设置同步资源值
                    if (zkLocklatch.getCount() <= 0) {
                        zkLocklatch = new CountDownLatch(1);
                    }
                    // 阻塞线程
                    zkLocklatch.await();
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    /**
     * 释放分布式锁
     */
    public boolean releaseLock() {
        try {
            if (client.checkExists().forPath("/" + ZK_LOCK_PROJECT + "/" + DISTRIBUTED_LOCK) != null) {
                client.delete().forPath("/" + ZK_LOCK_PROJECT + "/" + DISTRIBUTED_LOCK);
            }
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        log.info("分布式锁释放完毕");
        return true;
    }


    /**
     * 创建watcher监听
     */
    public void addWatcherToLock(String path) throws Exception {
        final PathChildrenCache cache = new PathChildrenCache(client, path, true);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                    String path = event.getData().getPath();
                    log.info("上一个会话已释放或该会话已断开，节点路径为:" + path);
                    if (path.contains(DISTRIBUTED_LOCK)) {
                        log.info("释放计数器,让当前请求获得分布式锁...");
                        zkLocklatch.countDown();
                    }
                }
            }
        });
    }


}
